package com.orangeandbronze.demos.mapreduce;

import com.google.appengine.tools.mapreduce.KeyValue;
import com.google.appengine.tools.mapreduce.Mapper;

public class IndexMapper extends Mapper<KeyValue<String, byte[]>, String, String> {

	private static final long serialVersionUID = -6882924941408407298L;

	@Override
	public void map(KeyValue<String, byte[]> data) {
		final String filename = data.getKey();
		final String line = new String(data.getValue());
		final String[] words = line.split("\\W+");
		for (String word : words) {
			if (!word.isEmpty()) {
				getContext().emit(word.toLowerCase(), filename);
			}
		}
	}

	/*
def index_map(data):
  """Index demo map function."""
  (entry, text_fn) = data
  text = text_fn()

  logging.debug("Got %s", entry.filename)
  for s in split_into_sentences(text):
    for w in split_into_words(s.lower()):
      yield (w, entry.filename)


def index_reduce(key, values):
  """Index demo reduce function."""
  yield "%s: %s\n" % (key, list(set(values)))
  	 */
}
